package com.study.mqtt.config;

import java.util.Map;

import javax.annotation.Resource;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.event.ContextClosedEvent;

import com.study.mqtt.consumer.TestMqttMessageConsumer;


@Configuration
@Import({MqttTemplate.class})
public class MqttClientLifecycleListener implements ApplicationRunner, ApplicationListener<ContextClosedEvent> {

	private static final Logger log = LoggerFactory.getLogger(MqttClientLifecycleListener.class);

	private final MqttClient client;
	private final MqttConnectOptions options;
	private final MqttProperties properties;
	@Resource
	private ApplicationContext applicationContext;

	MqttClientLifecycleListener(MqttClient client, MqttConnectOptions options, MqttProperties properties) {
		this.client = client;
		this.options = options;
		this.properties = properties;
	}

	@Override
	public void run(ApplicationArguments args) throws Exception {
		log.debug("Connecting to broker:{}", this.properties.getBroker());
		this.client.connect(this.options);
		log.debug("Mqtt broker Connected");
		this.applicationContext.getBeansWithAnnotation(MqttMessageListener.class).forEach(this::subscribe);
	}

	private void subscribe(String beanName, Object consumer) {
		if (consumer instanceof MqttMessageConsumer) {
			Class<?> clazz = AopProxyUtils.ultimateTargetClass(consumer);
			MqttMessageListener annotation = clazz.<MqttMessageListener>getAnnotation(MqttMessageListener.class);
			String topic = annotation.topic();
			int qos = annotation.qos();
			String shareTopicPrefix = annotation.shareTopicPrefix();
			if (topic.startsWith(shareTopicPrefix))
				topic = topic.substring(shareTopicPrefix.length());
//			MqttClient client = (MqttClient) this.applicationContext.getBean(MqttClient.class);
			try {
				client.subscribe(topic, qos);
				MqttConfig.CONSUMERS.put(topic, (MqttMessageConsumer)consumer);
			} catch (MqttException e) {
				log.error("The subscription failed :{}", topic, e);
			}
		}
	}

	@Override
	public void onApplicationEvent(ContextClosedEvent event) {
		log.debug("Disconnecting from broker:{}", this.properties.getBroker());
		try {
			this.client.disconnectForcibly();
			log.debug("Disconnected");
		} catch (MqttException e) {
			log.error(e.getMessage(), (Throwable) e);
		}
	}

}
